package com.example.demo.task;

import com.example.demo.bean.KafkaLog;
import com.example.demo.controller.SendMessageController;
import com.example.demo.mapper.KafkaLogMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.CollectionUtils;

import java.util.List;

@Configuration
public class KafkaTask {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private static final int limitNum = 3;
    private static final String sendTryFlag = "1";
    @Autowired
    KafkaLogMapper kafkaLogMapper;

    @Autowired
    SendMessageController sendMessageController;

    @Scheduled(cron = "0 */1 * * * ?")
    private void failTask() {
        //redis 分布式锁
        List<KafkaLog> kafkaLogList = kafkaLogMapper.queryKafkaLogFailByStatus();
        if (CollectionUtils.isEmpty(kafkaLogList)) {
            logger.info("kafkaTask 没有失败记录");
            return;
        }
        for (KafkaLog log : kafkaLogList) {
            if (sendTryFlag.equals(log.getSend())) {
                sendMessageController.send(log.getMessageBody(), log.getMessageId());
                continue;
            }
            if (Integer.parseInt(log.getNum()) > limitNum) {
                logger.info("kafkaTask 超过限制次数:{}", log.getMessageId());
                continue;
            }
            sendMessageController.send(log.getMessageBody(), log.getMessageId());
        }
        logger.info("kafkaTask 执行成功");
    }

    @Scheduled(cron = "0 */2 * * * ?")
    private void timeOutTask() {
        //redis 分布式锁
        //超时时间要根据具体的消息积压的可能时间判断，或者不做该定时任务，交由人工处理
        List<KafkaLog> kafkaLogList = kafkaLogMapper.queryKafkaLogFailByTimeout();
        if (CollectionUtils.isEmpty(kafkaLogList)) {
            logger.info("kafkaTask 没有消费超时记录");
            return;
        }
        for (KafkaLog log : kafkaLogList) {
            if (sendTryFlag.equals(log.getSend())) {
                sendMessageController.send(log.getMessageBody(), log.getMessageId());
                continue;
            }
            if (Integer.parseInt(log.getNum()) > limitNum) {
                logger.info("kafkaTask 超过限制次数:{}", log.getMessageId());
                continue;
            }
            sendMessageController.send(log.getMessageBody(), log.getMessageId());
        }
        logger.info("kafkaTask 执行成功");
    }
}
